package com.gaofei.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class KafkaConsumer {

    @Autowired
    private RedisTemplate<String,String> redisTemplate;

    @KafkaListener(topics = "2105A" ,containerFactory="ackContainerFactory")
    public void revMsg(ConsumerRecord<String, String> msgObject, Acknowledgment ack ){

        String key = msgObject.key();
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(key, null,3, TimeUnit.SECONDS);//10s过期
        //看门狗,分布式锁的框架---Redission
        if(aBoolean){
            String msg = msgObject.value();
            System.out.println("收到了消息:"+key+":"+msg);//消费消息
            //拿到消息根据消息来处理业务1
            //出异常了
            //拿到消息根据消息来处理业务2

            ////告诉kafka 的broker 节点  消费成功了.你可以移动指针了.
            ack.acknowledge();
        }else {
            System.out.println("重复消费了!!!!!");
        }



    }
}
